package com.diy.sigmund.diyorder.common.middleware.mq.rocketmq.spring.one;

import com.diy.sigmund.diycommon.entity.base.exception.DiyRuntimeException;
import com.diy.sigmund.diycommon.test.entity.MqTestVO;
import com.diy.sigmund.diycommon.util.json.JacksonUtil;
import java.util.Objects;
import javax.annotation.Resource;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * <br>普通消费
 * <br>实现RocketMQListener接口
 * <br>申明泛型对象即业务对象
 * <br>使用@RocketMQMessageListener注解
 * <br>配置@RocketMQMessageListener注
 * <br>必须配置项入下
 * <br>group
 * <br>topic
 * <br>使用@Component注解或者子注解
 * <br>实现接口方法
 *
 * @author ylm-sigmund
 * @since 2022/5/5 22:21
 */
@Component
@RocketMQMessageListener(
        // 要求相同角色的消费者拥有完全相同的订阅和消费者组才能正确实现负载均衡。它是必需的，并且必须是全球唯一的。
        consumerGroup = "${group.consumer.sendDataType}",
        topic = "${topic.sendDataType}",
        // 控制如何选择消息。也可以看看：SelectorType
        selectorType = SelectorType.TAG,
        // 控制可以选择的消息。语法请参见SelectorType.TAG和SelectorType.SQL92
        selectorExpression = "*",
        // 控制消费模式，可选择并发或有序接收消息。
        consumeMode = ConsumeMode.CONCURRENTLY,
        // 控制消息模式，如果希望所有订阅者都接收消息，广播是一个不错的选择。
        messageModel = MessageModel.CLUSTERING,
        // 消费者线程号。
        consumeThreadNumber = 20,
        // 最大重复消费次数，与delayLevel类似。在并发模式下，-1 表示 16；在有序模式下，-1 表示 Integer.MAX_VALUE。
        maxReconsumeTimes = 2,
        // 一条消息可能阻塞消费线程的最长时间（以分钟为单位）
        consumeTimeout = 15L,
        // 发送回复消息超时。
        replyTimeout = 3000,
        // “访问密钥”的属性。
        accessKey = RocketMQMessageListener.ACCESS_KEY_PLACEHOLDER,
        // “密钥”的属性。
        secretKey = RocketMQMessageListener.SECRET_KEY_PLACEHOLDER,
        // 消息跟踪的切换标志实例。
        enableMsgTrace = false,
        // 消息跟踪主题的名称值。如果不配置，可以使用默认的跟踪主题名称。
        customizedTraceTopic = RocketMQMessageListener.TRACE_TOPIC_PLACEHOLDER,
        nameServer = RocketMQMessageListener.NAME_SERVER_PLACEHOLDER,
        // “访问通道”的属性。
        accessChannel = RocketMQMessageListener.ACCESS_CHANNEL_PLACEHOLDER,
        // “tlsEnable”的属性默认为false。
        tlsEnable = "false",
        // 消费者的命名空间。
        namespace = "",
        // 并发模式下的消息消费重试策略。 -1，不重试，直接放入DLQ ；0，broker控制重试频率；>0，client控制重试频率
        // 注：只有在消息模式为MessageModel.CLUSTERING集群模式时，Broker才会自动进行重试，广播消息是不会重试的。
        delayLevelWhenNextConsume = 2,
        // 在有序模式下暂停拉取的时间间隔，以毫秒为单位。最小值为 10，最大值为 30000。
        suspendCurrentQueueTimeMillis = 1000,
        // 关闭消费者时等待消息消费的最长时间，以毫秒为单位。最小值为 0
        awaitTerminationMillisWhenShutdown = 1000
)
public class SendDataTypeConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {

    private static final Logger log = LoggerFactory.getLogger(SendDataTypeConsumer.class);

    @Resource
    private RocketMQTemplate rocketMQTemplate;
    @Value("${topic.sendDataType.DLQ}")
    private String sendDataTypeDLQTopic;

    /**
     * <br>注意：Consumer里抛出异常才会重试，所以使用者不要把Consumer里的整个代码try-catch
     */
    @Override
    public void onMessage(MessageExt messageExt) {
        int reconsumeTimes = messageExt.getReconsumeTimes();
        log.info("reconsumeTimes:{}", reconsumeTimes);
        MqTestVO mqTestVO = JacksonUtil.toObject(new String(messageExt.getBody()), MqTestVO.class);
        if (Objects.equals("111", mqTestVO.getUserId())) {
            if (reconsumeTimes >= 2) {
                log.info("将该消息投入死信队列");
                rocketMQTemplate.convertAndSend(sendDataTypeDLQTopic + ":*", mqTestVO, null, null);
            }
            throw new DiyRuntimeException();
        }
        log.info(
                String.format("SendDataTypeConsumer name is %s data is %s", this.getClass().getSimpleName(), mqTestVO));
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        // 从消息队列头部开始消费会导致本地测试时不消费，重启项目时才消费，故使用默认配置
        /**
         * 消费者启动的消费点。
         *
         * 消费点分为三点：
         * CONSUME_FROM_LAST_OFFSET ：消费者客户端从之前停止的地方开始。如果是新启动的消费者客户端，根据消费者群体的老化情况，有两种情况：
         * 如果consumer group是最近才创建的，最早订阅的消息还没有过期，说明consumer group代表的是最近上线的业务，从头开始消费；
         * 如果最早订阅的消息已过期，则从最新消息开始消费，这意味着在启动时间戳之前生成的消息将被忽略。
         * CONSUME_FROM_FIRST_OFFSET ：消费者客户端将从最早可用的消息开始。
         * CONSUME_FROM_TIMESTAMP ：Consumer 客户端将从指定的时间戳开始，这意味着在consumeTimestamp之前生成的消息将被忽略
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 一条消息可能阻塞消费线程的最长时间（以分钟为单位）。设置消费超时时间(分钟)
        consumer.setConsumeTimeout(15);
        log.info("consumer:{}", consumer);
    }
}
